package ProcessFunctionTest

import Source.SensorReading
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object KeyedProcessFunction {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)


    val inputStream = env.socketTextStream("localhost", 7777)

    //转换成样例类
    val dataStream = inputStream
      .map(data => {
        val arr = data
          .split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }).keyBy(_.id)
      .process(new MyKeyedProcessFunction)

    env.execute()
  }
}

/**
 * 三个参数是：key的数据类型
 * 输入的数据类型
 * 输出的数据类型
 */
class MyKeyedProcessFunction extends KeyedProcessFunction[String, SensorReading, String] {

  var myState: ValueState[Int] = _

  override def open(parameters: Configuration): Unit = {
    myState = getRuntimeContext.getState(new ValueStateDescriptor[Int]("valueState", classOf[Int]))
  }

  override def processElement(i: SensorReading, context: KeyedProcessFunction[String, SensorReading, String]#Context,
                              collector: Collector[String]): Unit = {
    //获取当前的key，其实还可以直接从数据里面获取
    context.getCurrentKey
    //获取当前数据的时间戳
    context.timestamp()
    //获取当前的watermark
    context.timerService().currentWatermark()

    /**
     * 定义一个一分钟后触发的定时器，当定时器触发后，会执行onTimer方法
     * 可以注册多个定时器，定时器的区别就是时间戳
     * 不同的定时器执行时都是在onTimer方法里面
     */
    context.timerService().registerEventTimeTimer(context.timestamp() + 60 * 1000l)

    /**
     * 定时器的删除，删除也是传入一个参数，参数就是时间戳
     */
    //    context.timerService().deleteEventTimeTimer()
  }

  /**
   *
   * @param timestamp是定时器触发的时间 ,我们可以将根据不同的时间戳来判断不同的定时器，然后执行不一样的方法
   * @param ctx
   * @param out
   */
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext
                       , out: Collector[String]): Unit = {
  }
}